package com.atguigu.flink.datastreamapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Created by Smexy on 2022/12/13
 *
 *      分流。
 *          把一个流分为多个流。
 *                  1个主流，剩下的都是测流(SideOutput)
 *
 *
 *
 *       把s1类型的传感器，分入主流。
 *       s2到测流
 *       s3到测流
 */
public class Demo11_Process3
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //常见的数据类型，直接使用 Types.XXX 获取 TypeInformation
        // String:Types.STRING
        // 自定义的类型，可以使用 TypeInformation.of(Class c)
        OutputTag<WaterSensor> s2OT = new OutputTag<>("s2", TypeInformation.of(WaterSensor.class) );
        OutputTag<WaterSensor> s3OT = new OutputTag<>("s3",TypeInformation.of(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .process(new ProcessFunction<WaterSensor, WaterSensor>()
            {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                    String id = value.getId();

                    if ("s1".equals(id)) {
                        //主流输出
                        out.collect(value);
                    } else if ("s2".equals(id)) {
                               /*
                                    侧流输出

                                    output(OutputTag<X> outputTag, X value)
                                            OutputTag<X> outputTag: 一个标记，用于说明测流中输出的数据类型，为测流起一个id
                                             X value； 输出的数据
                                */
                        ctx.output(s2OT, value);
                    } else if ("s3".equals(id)) {
                        ctx.output(s3OT, value);
                    }

                }
            });


        //选择主流进行操作
        ds.print("主流");
        //获取侧流
        ds.getSideOutput(s2OT).printToErr("s2侧流");
        ds.getSideOutput(s3OT).printToErr("s3侧流");


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
